package org.ecbox.android.sdk.internal.stream;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.ecbox.android.sdk.EopHashMap;
import org.ecbox.android.sdk.EopUtils;
import org.ecbox.android.sdk.internal.stream.connect.ConnectionLifeCycleListener;
import org.ecbox.android.sdk.internal.stream.connect.HttpClient;
import org.ecbox.android.sdk.internal.stream.connect.HttpResponse;
import org.ecbox.android.sdk.internal.stream.message.EopCometMessageListener;
import org.ecbox.android.sdk.internal.stream.message.StreamMessageConsume;
import org.ecbox.android.sdk.internal.stream.message.StreamMsgConsumeFactory;
import org.ecbox.android.sdk.internal.util.RequestParametersHolder;
import org.ecbox.android.sdk.internal.util.StringUtils;

import android.util.Log;

/**
 * 异步消息处理程序
 * 
 * File: EopCometStreamImpl.java<br/>
 * Description: <br/>
 * 
 * Copyright: Copyright (c) 2012 ecbox.com<br/>
 * Company: ECBOX,Inc.<br/>
 * 
 * @author chenxiaochun
 * @date Dec 20, 2012
 * @version 1.0
 */
public class EopCometStreamImpl implements EopCometStream {

  private String log = EopCometStreamImpl.class.getName();

  private ConnectionLifeCycleListener connectionListener;
  private EopCometMessageListener cometMessageListener;
  private Configuration conf;
  private StreamMsgConsumeFactory msgConsumeFactory = null;

  /**
   * 停掉全部连接
   */
  private boolean gloableStop = false;
  private List<ControlThread> controlThreads = new ArrayList<ControlThread>();

  protected EopCometStreamImpl(Configuration conf) {
    this.conf = conf;
  }

  @Override
  public void setConnectionListener(ConnectionLifeCycleListener connectionLifeCycleListener) {
    this.connectionListener = connectionLifeCycleListener;
  }

  @Override
  public void setMessageListener(EopCometMessageListener cometMessageListener) {
    this.cometMessageListener = cometMessageListener;
  }

  @Override
  public void start() {
    Set<EopCometStreamRequest> cometRequests = conf.getConnectReqParam();
    for (EopCometStreamRequest cometRequest : cometRequests) {
      if (cometRequest.getConnectListener() == null) {
        cometRequest.setConnectListener(connectionListener);
      }
      if (cometRequest.getMsgListener() == null) {
        if (cometMessageListener == null) {// 如果又没有自己的message
          // listener又没设置全局的message
          // listener 则系统不能启动，抛错。
          throw new RuntimeException("Comet message listener must not null");
        }
        cometRequest.setMsgListener(cometMessageListener);
      }
    }
    msgConsumeFactory = new StreamMsgConsumeFactory(conf.getMinThreads(), conf.getMaxThreads(), conf.getQueueSize(),
        conf.getMsgConsumeThreadPoolRejectHandler());

    for (EopCometStreamRequest cometRequest : cometRequests) {
      // 这里不把线程设置成daemon线程的原因，如果设置成daemon线程后，如果主线程退出后这些线程都将自动退出
      ControlThread ct = new ControlThread(cometRequest, conf);
      controlThreads.add(ct);
      Thread t = new Thread(ct, "stream-control-thread-connectid-" + cometRequest.getConnectionName());
      t.setDaemon(false);
      t.start();
    }
  }

  /**
   * 在运行期添加新的长连接
   * 
   * @param newClient
   */
  @Override
  public void addNewStreamClient(EopCometStreamRequest newClient) {
    if (newClient.getConnectListener() == null) {
      newClient.setConnectListener(connectionListener);
    }
    if (newClient.getMsgListener() == null) {
      newClient.setMsgListener(cometMessageListener);
    }
    ControlThread ct = new ControlThread(newClient, conf);
    controlThreads.add(ct);
    Thread t = new Thread(ct, "stream-control-thread-connectid-" + newClient.getConnectionName());
    t.setDaemon(false);
    t.start();
  }

  /**
   * 关闭长连接。 这里没有休眠一段时间，让服务端清理资源， 这样的话客户端只要保证每次发起连接请求的connectid是一样的就不会出现问题。
   * 其实在客户端使用的过程中，客户端使用固定的connectid，在多连接情况下更应该 使用固定的connectid，并且不能超过服务端的最大连接数限制.
   */
  @Override
  public void stop() {
    Log.w(log, "start stop stream consume...");
    // 设置所有的线程都关闭掉
    gloableStop = true;
    // 关闭控制线程
    // 这里先复制一份controlThreads，因为在关闭的时候，control thread会从list中移除掉自己。
    Object[] copyControlThreads = controlThreads.toArray();
    for (Object o : copyControlThreads) {
      if (o instanceof ControlThread) {
        ControlThread ct = (ControlThread) o;
        try {
          ct.lock.lock();
          ct.controlCondition.signalAll();
        } catch (Exception e) {
          Log.e(log, e.getMessage());
        } finally {
          ct.lock.unlock();
        }
      }
    }
    // 关闭处理数据的线程池
    msgConsumeFactory.shutdown();
    Log.w(log, "stream consume stoped...");
  }

  /**
   * 连接管理线程
   * 
   * File: EopCometStreamImpl.java<br/>
   * Description: <br/>
   * 
   * Copyright: Copyright (c) 2012 ecbox.com<br/>
   * Company: ECBOX,Inc.<br/>
   * 
   * @author chenxiaochun
   * @date Dec 20, 2012
   * @version 1.0
   */
  public class ControlThread implements Runnable {

    private static final String threadName = "eop-stream-consume-thread";
    private TopCometStreamConsume currentStreamConsume;
    private boolean isReconnect = false;// 是否客户端发起重连

    private String serverRespCode = StreamConstants.CLIENT_FIRST_CONNECT;
    private ReentrantLock lock = new ReentrantLock();
    private Condition controlCondition = lock.newCondition();

    // 发生了多少次readtimeout
    private int readTimeoutOccureTimes = 0;
    private long lastStartConsumeThread = System.currentTimeMillis();

    // 听掉
    private boolean stop = false;

    // 请求
    private EopCometStreamRequest cometReq;
    private Configuration conf;

    public ControlThread(EopCometStreamRequest cometReq, Configuration conf) {
      this.cometReq = cometReq;
      this.conf = conf;
    }

    public ReentrantLock getLock() {
      return lock;
    }

    public Condition getControlCondition() {
      return controlCondition;
    }

    public void setServerRespCode(String serverRespCode) {
      this.serverRespCode = serverRespCode;
    }

    @Override
    public void run() {
      long lastSleepTime = 0;
      while (!stop) {
        if (gloableStop) {
          break;
        }
        try {
          if (StreamConstants.SERVER_DEPLOY.equals(serverRespCode)) {// 服务端在发布
            Log.d(log, "Server is upgrade sleep " + conf.getSleepTimeOfServerInUpgrade() + " seconds");
            try {
              Thread.sleep(conf.getSleepTimeOfServerInUpgrade() * 1000);
            } catch (InterruptedException e) {
              // ignore;
            }
            startConsumeThread();
          } else if (

          /* 客户端第一次发起连接请求 */
          StreamConstants.CLIENT_FIRST_CONNECT.equals(serverRespCode) ||

          /* 服务端主动断开了所有的连接 */
          StreamConstants.SERVER_REHASH.equals(serverRespCode) ||

          /* 连接到达最大时间 */
          StreamConstants.CONNECT_REACH_MAX_TIME.equals(serverRespCode) ||

          /* 在一些异常情况下需要重连 */
          StreamConstants.RECONNECT.equals(serverRespCode)) {
            startConsumeThread();
          } else if (

          /* 客户端自己把自己踢开 */
          StreamConstants.CLIENT_KICKOFF.equals(serverRespCode) ||

          /* 服务端把客户端踢开 */
          StreamConstants.SERVER_KICKOFF.equals(serverRespCode)) {
            if ((StreamConstants.CLIENT_KICKOFF.equals(serverRespCode) && !isReconnect)
                || StreamConstants.SERVER_KICKOFF.equals(serverRespCode)) {
              stop = true;
              if (currentStreamConsume != null) {
                currentStreamConsume.closed = true;
              }
              break;// 终止掉当前线程
            }
          } else {
            // 错误码设置出错，停止线程
            stop = true;
            break;
          }

          // 连接成功，开始休眠
          if (!stop && !gloableStop) {
            try {
              lock.lock();
              lastSleepTime = System.currentTimeMillis();
              controlCondition.await(conf.getHttpReconnectInterval(), TimeUnit.SECONDS);
              if (System.currentTimeMillis() - lastSleepTime >= (conf.getHttpReconnectInterval() - 5 * 60) * 1000) {
                /*
                 * 快要到达连接的最大时间了，需要重新发起连接
                 */
                serverRespCode = StreamConstants.RECONNECT;
                isReconnect = true;
              }// 否则，是由于某种原因被唤醒的
            } catch (Exception e) {
              Log.e(log, e.getMessage());
            } finally {
              lock.unlock();
            }
          }
        } catch (Throwable e) {
          Log.e(log, "Occur some error,stop the stream consume", e);
          stop = true;
          try {
            lock.lock();
            controlCondition.signalAll();
          } catch (Exception ex) {
            // ignore
          } finally {
            lock.unlock();
          }
        }
      }

      // 关闭资源
      if (currentStreamConsume != null) {
        currentStreamConsume.close();
      }

      // 此控制线程由于某种原因退出，从列表中移除掉
      controlThreads.remove(this);
    }

    /**
     * 开启消费线程
     * 
     */
    private void startConsumeThread() {
      StreamMessageConsume stream = null;
      try {
        stream = getMsgStreamImpl();
      } catch (Exception ex) {
        stop = true;
        gloableStop = true;
        if (cometReq.getConnectListener() != null) {
          cometReq.getConnectListener().onException(ex);
        }
        return;
      }
      currentStreamConsume = new TopCometStreamConsume(stream, this, cometReq.getConnectListener());
      Thread consumeThread = new Thread(currentStreamConsume, threadName);
      consumeThread.setDaemon(true);
      consumeThread.start();
      lastStartConsumeThread = System.currentTimeMillis();
      /**
       * 重新连接以后，修改isReconnect=false 这里去掉，因为当消息量很大的时候会出现：
       * 客户端重连先连接成功了，但是服务端在之前的连接上发送的client kickoff消息还没有处理完，
       * 导致在收到这个消息的时候，在控制线程里面处理的时候因为isReconnect = false了， 导致控制线程认为是客户端踢出。
       * 
       * 这里去掉的问题就是：如果isv需要在别的地方部署同一个appkey+userid+connect_id的长连接
       * 的时候需要把之前的那个关闭掉，否则会出现永远互相踢的过程。
       */
      // isReconnect = false;
    }

    /**
     * 获取消息流
     * 
     * @return
     * @throws EopCometSysErrorException
     * @throws Exception
     */
    private StreamMessageConsume getMsgStreamImpl() throws EopCometSysErrorException, Exception {
      if (cometReq.getConnectListener() != null) {
        // 触发连接前监听程序
        cometReq.getConnectListener().onBeforeConnect();
      }

      // 主要参数
      EopHashMap param = new EopHashMap();
      param.put(StreamConstants.PARAM_APP_KEY, cometReq.getAppKey());
      param.put(StreamConstants.PARAM_INSTANCE_USER_ID, cometReq.getInstanceUserId());
      param.put(StreamConstants.PARAM_TIMESTAMP, System.currentTimeMillis() / 1000);

      // 其他参数
      EopHashMap otherParam = new EopHashMap();
      if (!StringUtils.isEmpty(conf.getSignMethod())) {
        otherParam.put(StreamConstants.PARAM_SIGN_METHOD, conf.getSignMethod());
      }
      if (!StringUtils.isEmpty(cometReq.getAccessToken())) {
        otherParam.put(StreamConstants.PARAM_ACCESS_TOKEN, cometReq.getAccessToken());
      }
      if (!StringUtils.isEmpty(cometReq.getConnectionName())) {// 担心本地ip也获取不到
        otherParam.put(StreamConstants.PARAM_CONNECTION_NAME, cometReq.getConnectionName());
      }
      if (otherParam != null) {
        Iterator<Entry<String, String>> it = otherParam.entrySet().iterator();
        while (it.hasNext()) {
          Entry<String, String> e = it.next();
          param.put(e.getKey(), e.getValue());
        }
      }

      // 参数处理器
      RequestParametersHolder paramHolder = new RequestParametersHolder();
      paramHolder.setProtocalMustParams(param);
      String sign = null;
      try {
        sign = EopUtils.signEopRequestNew(paramHolder, cometReq.getSecret(), false);
        if (StringUtils.isEmpty(sign)) {
          throw new RuntimeException("Get sign error");
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      param.put(StreamConstants.PARAM_SIGN, sign);// 加入签名参数

      HttpClient httpClient = new HttpClient(conf, param);
      HttpResponse response = httpClient.post();

      return new StreamMessageConsume(msgConsumeFactory, response, cometReq.getMsgListener(), this);
    }
  }

  /**
   * 从长连接通道中读取消息
   * 
   * File: EopCometStreamImpl.java<br/>
   * Description: <br/>
   * 
   * Copyright: Copyright (c) 2012 ecbox.com<br/>
   * Company: ECBOX,Inc.<br/>
   * 
   * @author chenxiaochun
   * @date Dec 20, 2012
   * @version 1.0
   */
  class TopCometStreamConsume implements Runnable {

    private StreamMessageConsume stream;
    private boolean closed = false;
    private ControlThread ct;
    private ConnectionLifeCycleListener connectListener;

    TopCometStreamConsume(StreamMessageConsume stream, ControlThread ct, ConnectionLifeCycleListener connectListener) {
      this.stream = stream;
      this.ct = ct;
      this.connectListener = connectListener;
    }

    @Override
    public void run() {
      while (!gloableStop && !closed && stream.isAlive()) {
        try {
          stream.nextMsg();
        } catch (IOException e) {// 出现了read time out异常
          // 资源清理
          if (stream != null) {
            try {
              stream.close();
            } catch (IOException e1) {
              // ignore;
            }
          }
          stream = null;
          closed = true;
          /**
           * 30分钟内发送了10次read time out exception 告知isv，但是不退出，继续发起重新连接
           */
          ct.readTimeoutOccureTimes++;
          if (System.currentTimeMillis() - ct.lastStartConsumeThread < 30 * 60 * 1000) {
            // 如果30分钟内，read time out发生了10次，则告知isv
            if (ct.readTimeoutOccureTimes >= 10) {
              if (connectListener != null) {
                try {
                  connectListener.onMaxReadTimeoutException();
                } catch (Exception ex) {
                }
              }
              // 告知完后清0
              ct.readTimeoutOccureTimes = 0;
            }
          } else {// 如果超过30分钟，则清0
            ct.readTimeoutOccureTimes = 0;
          }
          // 通知重连
          ct.serverRespCode = StreamConstants.RECONNECT;
          try {
            ct.lock.lock();
            ct.controlCondition.signalAll();
          } catch (Exception e2) {
          } finally {
            ct.lock.unlock();
          }
        }
      }
      // 出现异常情况下做资源清理
      close();
    }

    /**
     * 在外面调用stop需要关闭整个长连接的时候，ControlThread会调用此方法，把一些资源正确的释放掉
     */
    public void close() {
      if (this.stream != null) {
        try {
          this.stream.close();
        } catch (Exception e) {
          // logger.warn(e.getMessage(), e);
          // ignore;
        }
      }
    }
  }

}
